#include "zrpc/net/event_loop.h"

#include "zrpc/base/log.h"

namespace zrpc {
#ifdef __linux__
int CreateEventfd() { return ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); }
#endif


#ifdef _WIN64
EventLoop::EventLoop()
    : thread_id_(std::this_thread::get_id()),
      epoller_(new Select(this)),
      op_(socket::Pipe(wakeup_fd_)),
      wakeup_channel_(new Channel(this, wakeup_fd_[0])),
      timer_queue_(new TimerQueue(this)),
      current_active_channel_(nullptr),
      running_(false),
      event_handling_(false),
      calling_pending_functors_(false) {
  wakeup_channel_->SetReadCallback(std::bind(&EventLoop::HandleRead, this));
  wakeup_channel_->EnableReading();
}
#endif

#ifdef __linux__
EventLoop::EventLoop()
    : thread_id_(std::this_thread::get_id()),
      wakeup_fd_(CreateEventfd()),
      epoller_(new Epoll(this)),
      timer_queue_(new TimerQueue(this)),
      wakeup_channel_(new Channel(this, wakeup_fd_)),
      current_active_channel_(nullptr),
      running_(false),
      event_handling_(false),
      calling_pending_functors_(false) {
  wakeup_channel_->SetReadCallback(std::bind(&EventLoop::HandleRead, this));
  wakeup_channel_->EnableReading();
}
#endif

#ifdef __APPLE__
EventLoop::EventLoop()
    : thread_id_(std::this_thread::get_id()),
      epoller_(new Poll(this)),
      op_(socketpair(AF_UNIX, SOCK_STREAM, 0, wakeup_fd_)),
      wakeup_channel_(new Channel(this, wakeup_fd_[1])),
      timer_queue_(new TimerQueue(this)),
      current_active_channel_(nullptr),
      running_(false),
      event_handling_(false),
      calling_pending_functors_(false) {
  wakeup_channel_->SetReadCallback(std::bind(&EventLoop::HandleRead, this));
  wakeup_channel_->EnableReading();
}
#endif

void EventLoop::AbortNotInLoopThread() { assert(false); }

EventLoop::~EventLoop() {
  wakeup_channel_->DisableAll();
  wakeup_channel_->Remove();
#ifdef __linux__
  socket::Close(wakeup_fd_);
#else
  socket::Close(wakeup_fd_[0]);
  socket::Close(wakeup_fd_[1]);
#endif
}

void EventLoop::AssertInLoopThread() {
  if (!IsInLoopThread()) {
    AbortNotInLoopThread();
  }
}

void EventLoop::HandlerTimerQueue() { timer_queue_->HandleRead(); }

bool EventLoop::IsInLoopThread() const {
  return thread_id_ == std::this_thread::get_id();
}

bool EventLoop::GetEventHandling() const { return event_handling_; }

std::thread::id EventLoop::GetThreadId() const { return thread_id_; }

void EventLoop::UpdateChannel(Channel *channel) {
  assert(channel->OwnerLoop() == this);
  AssertInLoopThread();
  epoller_->UpdateChannel(channel);
}

void EventLoop::RemoveChannel(Channel *channel) {
  assert(channel->OwnerLoop() == this);
  AssertInLoopThread();
  if (event_handling_) {
    assert(current_active_channel_ == channel ||
           std::find(active_channels_.begin(), active_channels_.end(),
                     channel) == active_channels_.end());
  }
  epoller_->RemoveChannel(channel);
}

void EventLoop::CancelAfter(const std::shared_ptr<Timer> &timer) {
  timer_queue_->CancelTimer(timer);
}

std::shared_ptr<Timer> EventLoop::RunAfter(double when, bool repeat,
                                           TimerCallback &&cb) {
  return timer_queue_->AddTimer(when, repeat, std::move(cb));
}

std::shared_ptr<Timer> EventLoop::RunAt(TimeStamp &&stamp, double when,
                                        bool repeat, TimerCallback &&cb) {
  return timer_queue_->AddTimer(std::move(stamp), when, repeat, std::move(cb));
}

bool EventLoop::HasChannel(Channel *channel) {
  assert(channel->OwnerLoop() == this);
  AssertInLoopThread();
  return epoller_->HasChannel(channel);
}

void EventLoop::HandleRead() {
  uint64_t one = 1;
#ifdef __linux__
  ssize_t n = socket::Read(wakeup_fd_, &one, sizeof one);
#endif

#ifdef __APPLE__
  ssize_t n = socket::Read(wakeup_fd_[1], &one, sizeof one);
#endif

#ifdef _WIN64
  ssize_t n = socket::Read(wakeup_fd_[0], &one, sizeof one);
#endif
  assert(n == sizeof one);
}

void EventLoop::Quit() {
  running_ = false;
  if (!IsInLoopThread()) {
    Wakeup();
  }
}

void EventLoop::QuitInLoop() { RunInLoop(std::bind(&EventLoop::Quit, this)); }

void EventLoop::Wakeup() {
  uint64_t one = 1;
#ifdef __linux__
  ssize_t n = socket::Write(wakeup_fd_, &one, sizeof one);
  assert(n == sizeof one);
#endif

#ifdef __APPLE__
  ssize_t n = socket::Write(wakeup_fd_[0], &one, sizeof one);
  assert(n == sizeof one);
#endif

#ifdef _WIN64
  ssize_t n = socket::Write(wakeup_fd_[1], &one, sizeof(one));
  assert(n == sizeof one);
#endif
}

void EventLoop::RunInLoop(Functor &&cb) {
  if (IsInLoopThread()) {
    cb();
  } else {
    QueueInLoop(std::move(cb));
  }
}

void EventLoop::DelayRunInLoop(Functor &&cb) {
  {
    std::unique_lock<std::mutex> lk(mutex_);
    pending_functors_.push_back(std::move(cb));
  }

  Wakeup();
}

void EventLoop::QueueInLoop(Functor &&cb) {
  {
    std::unique_lock<std::mutex> lk(mutex_);
    pending_functors_.push_back(std::move(cb));
  }

  if (!IsInLoopThread() || calling_pending_functors_) {
    Wakeup();
  }
}

void EventLoop::DoPendingFunctors() {
  calling_pending_functors_ = true;

  {
    std::unique_lock<std::mutex> lk(mutex_);
    functors_.swap(pending_functors_);
  }

  for (size_t i = 0; i < functors_.size(); ++i) {
    functors_[i]();
  }

  functors_.clear();
  calling_pending_functors_ = false;
}

void EventLoop::Run() {
  running_ = true;
  while (running_) {
    active_channels_.clear();
    epoller_->EpollWait(&active_channels_);
    event_handling_ = true;

    for (auto &it : active_channels_) {
      current_active_channel_ = it;
      current_active_channel_->HandleEvent();
    }

    current_active_channel_ = nullptr;
    event_handling_ = false;
    DoPendingFunctors();
  }
}

int EventLoop::GetTimeOut() const {
  return timer_queue_->GetTimeout();
}

}  // namespace zrpc